import akshare as ak
import pymysql
from sqlalchemy import create_engine
import pandas as pd
import datetime
import time

'''使用目前所有郑期所的当期合约进行跨期套利的数据分析'''
#  使用AKSshare获取郑州商品交易所的所有当期合约信息
futures_contract_info_czce_df = ak.futures_contract_info_czce(date="20251010")  # 确保日期是交易日符合接口标准,若使用当天日期,要在收盘后
# print(futures_contract_info_czce_df.head())
# print(futures_contract_info_czce_df)


'''数据清洗'''
#  将获取的dataframe的索引列进行重命名
columns_to_keep = ['合约代码', '产品代码', '交易所MIC编码', '第一交易日',
                   '到期时间待国家公布2025年节假日安排后进行调整', '交易单位', '交易保证金率', '最小变动价位']
rename_dict = {'合约代码': 'instrumentID', '产品代码': 'instrumentClass', '交易所MIC编码': 'exchangeID',
               '第一交易日': 'listDate', '到期时间待国家公布2025年节假日安排后进行调整': 'expiryDate',
               '交易单位': 'volumeMutiplier', '交易保证金率': 'MarginRatio', '最小变动价位': 'priceTick'}
futures_contract_info_czce_df = futures_contract_info_czce_df[columns_to_keep].rename(columns=rename_dict)
#  对字符数据进行修改并删除冗余字符
futures_contract_info_czce_df['exchangeID'] = futures_contract_info_czce_df['exchangeID'].str.replace('X', 'C')
futures_contract_info_czce_df['expiryDate'] = futures_contract_info_czce_df['expiryDate'].str.replace(r'[北京时间]', '',
                                                                                                      regex=True)
futures_contract_info_czce_df['listDate'] = pd.to_datetime(futures_contract_info_czce_df['listDate'])
futures_contract_info_czce_df['expiryDate'] = pd.to_datetime(futures_contract_info_czce_df['expiryDate'])
earliest_listdate = futures_contract_info_czce_df['listDate'].min()  # 获取合约中最早的上市日期是20241015
earlist_expirydate = futures_contract_info_czce_df['expiryDate'].min()  # 获取合约中最早的退市日期20251015
# print(earlist_expirydate)
# print(earliest_listdate)
# print(futures_contract_info_czce_df)


'''尝试使用sqlalchemy的create_engine模块和把dataframe快速存储到mysql,注意这种方式插入的数据的类型是text'''
engine = create_engine('mysql+mysqlconnector://root:123456@localhost:3306/futuredata_schema?charset=utf8')
conn = engine.connect()
futures_contract_info_czce_df.to_sql(name='futuresinfo_czce', con=conn, index=False, if_exists='replace')

conn.close()  # 关闭连接

'''获取所有合约日线行情数据'''
get_futures_daily_df = ak.get_futures_daily(start_date="20241015", end_date="20251002", market="CZCE")
# print(get_futures_daily_df)
# 只保留当期合约的日线行情数据
valid_instruments = futures_contract_info_czce_df['instrumentID'].tolist()
filtered_daily = get_futures_daily_df[get_futures_daily_df['symbol'].isin(valid_instruments)]
# print(len(get_futures_daily_df))
# print(len(filtered_daily))
# 对列索引进行修改方便插入数据库
filtered_daily = filtered_daily.drop(['open_interest', 'turnover', 'variety'], axis=1)
filtered_daily = filtered_daily.rename(
    columns={'symbol': 'instrumentID', 'date': 'updateDate', 'open': 'openPrice', 'high': 'highPrice',
             'low': 'lowPrice', 'close': 'todayClosePrice', 'volume': 'todayTradeVolume', 'settle': 'todaySettlePrice',
             'pre_settle': 'preSettlePrice'})
# print(filtered_daily.columns)


# total_future_list = futures_contract_info_czce_df['instrumentID'].tolist()
# start_time = time.time()
# future_hist_dict = {}
# pending_future_list = list(set(total_future_list) - set(future_hist_dict))
# for num, instrumentID in enumerate(pending_future_list):
#     if num % 10 == 0:
#         print(f"has got {num} future data,achieve {num / len(pending_future_list)},cost {time.time() - start_time}")
#     tmp_df = filtered_daily[filtered_daily['instrumentID'] == instrumentID]
#     future_hist_dict[instrumentID] = tmp_df
# print(future_hist_dict['CY607'])


'''使用try-except语句进行数据库的初始连接'''
# conn = None
# try:
#     conn = pymysql.connect(host='localhost', user='root2', port=3306, password='123456', database='futuredata_schema', charset='utf8')
# except Exception as e:
#     print("异常：", e)
# finally:
#     if conn:
#         conn.close()


'''使用pymysql进行数据库的连接'''
mysqlconn = pymysql.connect(host='localhost', port=3306, user='root', password='123456', charset='utf8',
                            database='futuredata_schema')  # 创建连接
cursor = mysqlconn.cursor(cursor=pymysql.cursors.DictCursor)  # 创建游标

# 建立日线行情数据表
sql = '''
CREATE TABLE futuredaymarketdata_table(
id INT AUTO_INCREMENT PRIMARY KEY,
instrumentID CHAR(6) NOT NULL,
updateDate timestamp,
openPrice FLOAT,
highPrice FLOAT,
lowPrice FLOAT,
todayClosePrice FLOAT,
todayTradeVolume FLOAT,
todaySettlePrice FLOAT,
preSettlePrice FLOAT
)ENGINE=InnoDB DEFAULT CHARSET=utf8;
'''
cursor.execute('DROP TABLE IF EXISTS futuredaymarketdata_table;')
cursor.execute(sql)

# 批量插入数据
daybar_sql = "INSERT INTO futuredaymarketdata_table(instrumentID, updateDate, openPrice, highPrice, lowPrice,todayClosePrice, todayTradeVolume, todaySettlePrice, preSettlePrice)" \
             "values(%s,%s,%s,%s,%s,%s,%s,%s,%s)"

# 使用try语句进行执行sql语句
try:
    cursor.executemany(daybar_sql, filtered_daily.values.tolist())
    mysqlconn.commit()  # 提交事务
except Exception as e:
    print(e)
    mysqlconn.rollback()  # 若出现报错进行数据回滚

'''增量更新'''
add_futures_daily_df = ak.get_futures_daily(start_date="20251003", end_date='20251010', market='CZCE')
add_futures_daily_df = add_futures_daily_df.drop(['open_interest', 'turnover', 'variety'], axis=1)
add_futures_daily_df = add_futures_daily_df.rename(
    columns={'symbol': 'instrumentID', 'date': 'updateDate', 'open': 'openPrice', 'high': 'highPrice',
             'low': 'lowPrice', 'close': 'todayClosePrice', 'volume': 'todayTradeVolume', 'settle': 'todaySettlePrice',
             'pre_settle': 'preSettlePrice'})
print(add_futures_daily_df)
sql = 'INSERT INTO futuredaymarketdata_table(instrumentID, updateDate, openPrice, highPrice, lowPrice,todayClosePrice, todayTradeVolume, todaySettlePrice, preSettlePrice)' \
      'values(%s, %s, %s, %s, %s, %s, %s, %s, %s)'
try:
    cursor.executemany(sql, add_futures_daily_df.values.tolist())
    mysqlconn.commit()
except Exception as e:
    print(e)
    mysqlconn.rollback()

'''查询数据条数'''
sql = 'select count(*) from futuredaymarketdata_table;'
cursor.execute(sql)
result = cursor.fetchall()
print(result)

cursor.close()  # 关闭游标
mysqlconn.close()  # 关闭数据库连接
